package mr

import (
	"crypto/rand"
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"math/big"
	"net/rpc"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"
)

//worker has functions include map and reduce
//worker should send hearten message to server to register themselves
//if worker get response about "please exit" then finish process
//if worker get response about nothing don't do anything and keep send heaten is ok
//if worker get response about map task(here should be a file name)
//then handle input file to intermediate file named mr-X-Y
//X is the index of map task
//Y is the index of reduce task
//every key get hash key and % nReduce to decide to which Y should be output
//(for example if the hash key of "abc" is 3 and now the map task index of worker is 4
//the output file name should be mr-4-3)
//if worker get response about reduce task(here should be reduce task index ownY)
//then compute all files named mr-X-ownY and write output file named mr-out-ownY
//everytime finished task they should send "finished" message to server

// KeyValue
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

// ByKey for sorting by key.
type ByKey []KeyValue

// Len for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	// Your worker implementation here.
	workID, _ := rand.Int(rand.Reader, big.NewInt(100000000))

	// uncomment to send the Example RPC to the coordinator.
	// CallExample()
	taskType := "none"
	for taskType != "exit" {
		//每秒调用一次RPC
		time.Sleep(time.Second)
		args := AskingArgs{}
		args.WorkId = int(workID.Int64())
		args.AskingType = "asking"

		workID := int(workID.Int64())
		reply := Reply{}
		Call(args.WorkId, &args, &reply)
		log.Println("worker" + strconv.Itoa(workID) + "发送请求")
		log.Println("Worker" + strconv.Itoa(workID) + "响应任务类型为" + reply.TaskType)
		//maybe exit none map reduce
		taskType = reply.TaskType
		taskIndex := reply.TaskIndex
		if reply.TaskType == "map" {
			log.Println("Worker" + strconv.Itoa(workID) + "执行map task: " + reply.Filename + " ，index为" + strconv.Itoa(taskIndex))
			fileName := reply.Filename
			//读文件内容并且输出文件mr-X-Y，当完成时发送一次FIN请求RPC
			file, err := os.Open(fileName)
			if err != nil {
				log.Fatalf("cannot open %v", fileName)
			}
			content, err := ioutil.ReadAll(file)
			if err != nil {
				log.Fatalf("cannot read %v", fileName)
			}
			file.Close()
			log.Println("Worker" + strconv.Itoa(workID) + "读取文件成功，开始处理文件")
			//mapf will split content by " "
			//把文件内容根据key放入桶中
			kva := mapf(fileName, string(content))
			var intermediates [][]KeyValue
			for i := 0; i < reply.NReduce; i++ {
				intermediates = append(intermediates, []KeyValue{})
			}
			for _, value := range kva {
				hashKey := ihash(value.Key)
				intermediates[hashKey%reply.NReduce] = append(intermediates[hashKey%reply.NReduce], value)
			}
			for i := 0; i < reply.NReduce; i++ {
				sort.Sort(ByKey(intermediates[i]))
			}
			log.Println("Worker" + strconv.Itoa(workID) + "处理文件内容完成，输出intermediate文件")
			//输出到文件
			var oname string
			for i := 0; i < reply.NReduce; i++ {
				oname = "mr-" + strconv.Itoa(reply.TaskIndex) + "-" + strconv.Itoa(i)
				ofile, _ := os.Create(oname)
				data, _ := json.MarshalIndent(intermediates[i], "", "    ")
				fmt.Fprintf(ofile, "%s", data)
				ofile.Close()
			}

			//调用RPC传达完成task
			args := AskingArgs{}
			args.WorkId = workID
			args.AskingType = "finish"
			args.TaskType = reply.TaskType
			args.TaskIndex = taskIndex
			reply := Reply{}
			log.Println("Worker" + strconv.Itoa(workID) + "发送finish请求，任务类型为" + taskType + "，任务id为" + strconv.Itoa(taskIndex))
			Call(workID, &args, &reply)
		} else if reply.TaskType == "reduce" {
			log.Println("Worker" + strconv.Itoa(workID) + "执行reduce任务")
			reduceIndex := reply.TaskIndex
			pwd, _ := os.Getwd()

			log.Println("Worker" + strconv.Itoa(workID) + "扫描Y文件，Y为" + strconv.Itoa(reply.TaskIndex))
			//获取当前目录下的文件或目录名(包含路径)
			filepathNames, err := filepath.Glob(filepath.Join(pwd, "*"))
			if err != nil {
				log.Fatal(err)
			}
			var intermediatorfiles []string
			log.Println("Worker" + strconv.Itoa(workID) + "文件包含")
			for _, name := range filepathNames {
				sp := strings.Split(name, "/")
				filename := sp[len(sp)-1]
				//log.Println("文件：" + filename)
				if strings.HasPrefix(filename, "mr-") && strings.HasSuffix(filename, strconv.Itoa(reduceIndex)) {
					intermediatorfiles = append(intermediatorfiles, filename)
					log.Println("Worker" + strconv.Itoa(workID) + "加入intermediate文件：" + filename)
				}
			}
			log.Println("Worker" + strconv.Itoa(workID) + "开始处理intermediate文件内容")
			var intermediate []KeyValue
			for _, fileName := range intermediatorfiles {
				file, err := os.Open(fileName)
				if err != nil {
					log.Fatalf("cannot open %v", fileName)
				}
				content, err := ioutil.ReadAll(file)
				if err != nil {
					log.Fatalf("cannot read %v", fileName)
				}
				//把content反序列化加入intermediate
				var tmpkv []KeyValue
				json.Unmarshal(content, &tmpkv)
				log.Println("Worker" + strconv.Itoa(workID) + fileName + "文件中含有" + strconv.Itoa(len(tmpkv)) + "条KV")
				intermediate = append(intermediate, tmpkv...)
				file.Close()
			}
			log.Println("Worker" + strconv.Itoa(workID) + "intermediate内容读入内存开始排序")
			sort.Sort(ByKey(intermediate))
			log.Println("Worker" + strconv.Itoa(workID) + "开始输出out文件")
			//output
			oname := "mr-out-" + strconv.Itoa(reduceIndex)
			ofile, _ := os.Create(oname)

			i := 0
			log.Println("Worker" + strconv.Itoa(workID) + "reduce任务" + strconv.Itoa(reduceIndex) + "总共有" + strconv.Itoa(len(intermediate)) + "条KV")
			for i < len(intermediate) {
				j := i + 1
				for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
					j++
				}
				var values []string
				for k := i; k < j; k++ {
					values = append(values, intermediate[k].Value)
				}
				//reducef will return the length of values
				output := reducef(intermediate[i].Key, values)

				// this is the correct format for each line of Reduce output.
				fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

				i = j
			}

			ofile.Close()

			//调用RPC传达完成task
			args := AskingArgs{}
			args.WorkId = workID
			args.AskingType = "finish"
			args.TaskType = reply.TaskType
			args.TaskIndex = taskIndex
			reply := Reply{}
			log.Println("Worker" + strconv.Itoa(workID) + "发送finish请求，任务类型为" + taskType + "，任务id为" + strconv.Itoa(taskIndex))
			Call(workID, &args, &reply)
		}
	}
}
func Call(workId int, args *AskingArgs, reply *Reply) {

	// fill in the argument(s).
	args.WorkId = workId

	ok := call("Coordinator.Hearten", args, reply)
	if ok {
		log.Printf(reply.TaskType + " " + strconv.Itoa(reply.TaskIndex) + " " + reply.StartTime.String())
	} else {
		fmt.Printf("call failed!\n")
	}
}

//
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	sockname := coordinatorSock()
	log.Println("sockname is " + sockname)
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
